package com.hxq.demo4;

import org.apache.kafka.clients.producer.*;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/** 拦截器案例
 * @author hxq
 * @date 2022/4/18 16:20
 */
public class InterceptorProducer {

    public static void main(String[] args) throws InterruptedException {
        haveCallback();
    }

    private static void haveCallback() throws InterruptedException {

        Properties properties = setConfigInfo();
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //发送数据
        for (int i= 1; i<=4; i++){
            ProducerRecord<String, String> record = new ProducerRecord<>("first",
                    Integer.toString(i), "xxxxxxxxxx");
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e==null){
                        System.out.println("分区："+recordMetadata.partition()+"=====" + "游标："+recordMetadata.offset());
                    }else {
                        e.printStackTrace();
                    }

                }
            });
        }

        // 4 一定要关闭 producer，这样才会调用 interceptor 的 close 方法
        producer.close();

    }


    private static Properties setConfigInfo(){
        Properties properties = new Properties();
        //kafka集群
        properties.put("bootstrap.servers","212.129.155.35:9092");
        //ack
        properties.put("ack","all");
        //重试次数
        properties.put("retries",1);
        //批次大小  本版本是133  4条10字节的消息
        properties.put("batch.size",133);//只有数据积累到 batch.size 之后，sender线程才会发送数据
        //等待时间
        properties.put("linger.ms",10000);//如果数据迟迟未达到 batch.size，sender 等待 linger.time 之后就会发送数据 ms
        //RecordAccumulator 缓冲区大小
        properties.put("buffer.memory", 33554432);

        properties.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        //构建拦截链
        List<String> interceptors = new ArrayList<>();
        interceptors.add("com.hxq.interceptor.TimeInterceptor");
        interceptors.add("com.hxq.interceptor.CounterInterceptor");
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,interceptors);
        return properties;

    }
}
